package com.lagou.client;

import com.alibaba.fastjson.JSONObject;
import com.lagou.service.JSONSerializer;
import com.lagou.service.RpcDecoder;
import com.lagou.service.RpcEncoder;
import com.lagou.service.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer {

    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static UserClientHandler userClientHandler;
    
    public static List<MyChannel> myChannelList = new ArrayList<>();
    
    public static List<MyChannel> myChannelList2;
    
    public static Map<String, Long> startTimeMap = new HashMap<>();

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public Object createProxy(final Class<?> serviceClass,final RpcRequest rpcRequest){
        //借助JDK动态代理生成代理对象
        return  Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //（1）调用初始化netty客户端的方法
                if (userClientHandler == null) {
                    initClients();
                }
    
                if(myChannelList.size()>0) {
                    // 设置参数
                    userClientHandler.setPara(rpcRequest);
                    // 去服务端请求数据
                    executor.execute(new Runnable() {
                        @Override
                        public void run() {
                            
                            if(myChannelList.size()>0) {
                                
                                for(int i=0; i<myChannelList.size(); i++) {
                                    MyChannel mc = myChannelList.get(i);
                                    System.out.println(mc.host+":"+mc.port+"上次耗时"+mc.getLastTimeCosts());
                                }
                                
                                myChannelList.sort(new Comparator<MyChannel>() {
                                    @Override
                                    public int compare(MyChannel o1, MyChannel o2) {
                                        return (int) (o1.getLastTimeCosts()-o2.getLastTimeCosts());
                                    }
                                });
                                MyChannel mc1 = myChannelList.get(0);
                                System.out.println("选择"+mc1.host+":"+mc1.port+"上次耗时"+mc1.getLastTimeCosts());
                                Channel channel = mc1.getChannel();
                                startTimeMap.put(rpcRequest.getRequestId(), System.currentTimeMillis());
                                rpcRequest.setChannelId(mc1.getId());
                                channel.writeAndFlush(rpcRequest);
                            }
                        }
                    });
                }
                return null;
            }
        });


    }



    //2.初始化netty客户端
    public void initClient(String host, int port, List<MyChannel> myChannelList) throws InterruptedException {
         userClientHandler = new UserClientHandler();

        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast( new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(userClientHandler);
                    }
                });
    
        ChannelFuture channelFuture  = bootstrap.connect(host,port).sync();
        MyChannel myChannel = new MyChannel();
        myChannel.setHost(host);
        myChannel.setPort(port);
        myChannel.setChannel(channelFuture.channel());
        myChannel.setId(UUID.randomUUID().toString());
        myChannelList.add(myChannel);
    }
    
    public void initClients() throws Exception {
    
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                .retryPolicy(retryPolicy).sessionTimeoutMs(1000 * 6).connectionTimeoutMs(1000 * 6).namespace("rpc").build();
        client.start();
    
        List<String> childNodes = client.getChildren().forPath("/servers");
    
        for(String childNode : childNodes) {
            String addressStr = new String(client.getData().forPath("/servers/"+childNode));
            JSONObject addressJsonObject = JSONObject.parseObject(addressStr);
            initClient(addressJsonObject.getString("host"), addressJsonObject.getInteger("port"), myChannelList);
        }
    
        TreeCache treeCache = new TreeCache(client, "/servers");
    
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData eventData = event.getData();
                myChannelList2 = new ArrayList<>();
                List<String> childNodes2 = client.getChildren().forPath("/servers");
                for(String childNode : childNodes2) {
                    String addressStr = new String(client.getData().forPath("/servers/"+childNode));
                    JSONObject addressJsonObject = JSONObject.parseObject(addressStr);
                    initClient(addressJsonObject.getString("host"), addressJsonObject.getInteger("port"), myChannelList2);
                    myChannelList = myChannelList2;
                }
                
            }
        });
        treeCache.start();
    }


}
